can_enqueue? propagation detection that does not depend on Rails (#1672)
Karol Bucek
1 zmienionych plików z 9 dodań i 8 usunięć
+ 9
- 8
app/jobs/agent_propagate_job.rb
@@ -6,14 +6,15 @@ class AgentPropagateJob < ActiveJob::Base |
||
| 6 | 6 |
end |
| 7 | 7 |
|
| 8 | 8 |
def self.can_enqueue? |
| 9 |
- if Rails.configuration.active_job.queue_adapter == :delayed_job && |
|
| 10 |
- Delayed::Job.where(failed_at: nil, queue: 'propagation').count > 0 |
|
| 11 |
- return false |
|
| 12 |
- elsif Rails.configuration.active_job.queue_adapter == :resque && |
|
| 13 |
- (Resque.size('propagation') > 0 ||
|
|
| 14 |
- Resque.workers.select { |w| w.job && w.job['queue'] && w.job['queue']['propagation'] }.count > 0)
|
|
| 15 |
- return false |
|
| 9 |
+ case queue_adapter.name # not using class since it would load adapter dependent gems |
|
| 10 |
+ when 'ActiveJob::QueueAdapters::DelayedJobAdapter' |
|
| 11 |
+ return Delayed::Job.where(failed_at: nil, queue: 'propagation').count == 0 |
|
| 12 |
+ when 'ActiveJob::QueueAdapters::ResqueAdapter' |
|
| 13 |
+ return Resque.size('propagation') == 0 &&
|
|
| 14 |
+ Resque.workers.select { |w| w.job && w.job['queue'] && w.job['queue']['propagation'] }.count == 0
|
|
| 15 |
+ else |
|
| 16 |
+ raise NotImplementedError, "unsupported adapter: #{queue_adapter}"
|
|
| 16 | 17 |
end |
| 17 |
- true |
|
| 18 | 18 |
end |
| 19 |
+ |
|
| 19 | 20 |
end |